package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2WC {
  def main(args: Array[String]): Unit = {

    //spark 配置文件对象
    val conf = new SparkConf()


    //任务名
    conf.setAppName("wc")

    //指定spark 运行方法   local:本地运行(在单个JVM中运行)

    //提交到集群运行,master不能写,代码的优先级最高
    //conf.setMaster("local")


    //创建spark 上下文对象, 是spar的入口
    val sc = new SparkContext(conf)

    /**
      *
      * RDD : 弹性的分布式数据集
      * 代码层面可以当成一个List集合使用
      *
      */

    //1  读取文件
    //如果提交的集群,自动读取hdfs数据
    val lines: RDD[String] = sc.textFile("/data/words.txt")


    //2\ 将单词拆分
    val words: RDD[String] = lines.flatMap(line => line.split(","))


    //3 在每一单词后面加1
    val kv: RDD[(String, Int)] = words.map(word => (word, 1))

    /**
      * reduceByKey: 对每一个key所有的value进行聚合操作
      *
      */


    //4 统计单词的数量
    val count: RDD[(String, Int)] = kv.reduceByKey((x, y) => x + y)


    //5 整理数据
    val result: RDD[String] = count.map(kv => s"${kv._1}\t${kv._2}")


    //6 保存结果
    //如果在集群上运行,自动保存到hdfs
    result.saveAsTextFile("/data/wc1")

    /**
      *
      * 提交任务
      * spark-submit --class com.shujia.spark.Demo2WC --master yarn-client spark-1.0.jar
      *
      * 查看结果
      * hadoop dfs -cat  /data/wc1/
      *
      *
      **/

  }
}
